package org.apache.ocean.main.util;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.WeakHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService; 
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

public abstract class ArrayPool<T> {
	private Map<Pooled<T>, Internal> inUseMap = new HashMap<Pooled<T>, Internal>();
	private SortedMap<Integer, Set<Internal>> availableMap = new TreeMap<Integer, Set<Internal>>();
	private boolean isExactSize;
	private int initialSize;
	private long expiration;
	private ReentrantLock lock = new ReentrantLock();
	private static WeakHashMap<ArrayPool, Object> arrayPoolMap = new WeakHashMap<ArrayPool, Object>();
	private static ScheduledExecutorService reaperTimer;
	static {
		reaperTimer = Executors.newSingleThreadScheduledExecutor(); 
		reaperTimer.scheduleWithFixedDelay(new Reaper(), 1000 * 60, 1000 * 60, TimeUnit.MILLISECONDS);
	}

	public ArrayPool(int initialNum, int initialSize, boolean isExactSize, long expiration) {
		this.isExactSize = isExactSize;
		this.initialSize = initialSize;
		this.expiration = expiration;
		for (int x = 0; x < initialNum; x++) {
			get(initialSize);
		}
		arrayPoolMap.put(this, new Object());
	}

	private static class Reaper implements Runnable {
		public void run() {
			for (ArrayPool arrayPool : arrayPoolMap.keySet()) {
				arrayPool.removeExpired();
			}
		}
	}
	
	public void removeExpired() {
		lock.lock();
		try {
			for (Set<Internal> set : availableMap.values()) {
				for (Internal internal : set) {
					if (internal.isExpired(expiration)) {
						set.remove(internal);
					}
				}
			}
		} finally {
			lock.unlock();
		}
	}

	public class Internal {
		private Pooled<T> pooled;
		private long timestamp;

		public Internal(Pooled<T> pooled) {
			this.pooled = pooled;
			touch();
		}

		public Pooled<T> getPooled() {
			return pooled;
		}

		public void touch() {
			timestamp = Timeout.TIMER_THREAD.getTime();
		}

		public boolean isExpired(long duration) {
			return Timeout.TIMER_THREAD.getTime() > (duration + timestamp);
		}
	}

	public Pooled<T> get(int size) {
		lock.lock();
		try {
			Internal internal = null;
			if (isExactSize) {
				Set<Internal> set = availableMap.get(size);
				if (set.size() == 0) {
					internal = new Internal(createPooled(size));
				} else {
					internal = set.iterator().next();
				}
			} else {
				SortedMap<Integer, Set<Internal>> tailMap = availableMap.tailMap(size);
				if (tailMap.size() > 0) {
					Set<Internal> set = tailMap.get(tailMap.firstKey());
					if (set.size() > 0) {
						internal = set.iterator().next();
					} else {
						internal = new Internal(createPooled(size));
					}
				} else {
					internal = new Internal(createPooled(size));
				}
			}
			assert internal != null;
			availableMap.remove(internal);
			inUseMap.put(internal.getPooled(), internal);
			internal.touch();
			return internal.getPooled();
		} finally {
			lock.unlock();
		}
	}

	public static abstract class Pooled<T> {
		private ArrayPool<T> arrayPool;

		public Pooled(ArrayPool<T> arrayPool) {
			this.arrayPool = arrayPool;
		}

		public abstract T get();

		public abstract int size();

		public void finished() {
			arrayPool.finished(this);
		}
	}

	public abstract Pooled<T> createPooled(int size);

	public void finished(Pooled<T> pooled) {
		lock.lock();
		try {
			Internal internal = inUseMap.remove(pooled);
			internal.touch();
			Set<Internal> set = availableMap.get(pooled.size());
			if (set == null) {
				set = new HashSet<Internal>();
				availableMap.put(pooled.size(), set);
			}
			set.add(internal);
		} finally {
			lock.unlock();
		}
	}
}
